/*
 *@Project:
 *@Author: wanghong
 *@Date: 2019年07月09日
 *@Copyright: ©2019-2028 www.xfxb.net Inc. All rights reserved.
 */
package com.dopper.ws;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketHandshakeException;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import static io.netty.handler.codec.http.HttpMethod.GET;
import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;


/**
 * @author wanghong
 * @date 2019年07月09日
 */
public class WebSocketServerHandler extends ChannelInboundHandlerAdapter {

    private final static Logger log = LoggerFactory.getLogger(WebSocketServerHandler.class);

    private WebsocketServer websocketServer;

    WebSocketServerHandler(WebsocketServer websocketServer) {
        this.websocketServer = websocketServer;
    }

    private Map<String, String> parseHeaders(FullHttpRequest request) {
        if (request.headers() == null) {
            return Collections.emptyMap();
        }
        Map<String, String> headers = new HashMap<String, String>(request.headers().size());
        for (Entry<String, String> entry : request.headers()) {
            headers.put(entry.getKey(), entry.getValue());
        }
        return headers;
    }

    @Override
    public void channelRead(final ChannelHandlerContext channelHandlerContext, Object object){

        if (object instanceof FullHttpRequest) {
            FullHttpRequest request = (FullHttpRequest) object;
            if (!request.decoderResult().isSuccess()) {
                sendHttpResponse(channelHandlerContext, request,
                    new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
                ReferenceCountUtil.release(object);
                return;
            }
            if (!GET.equals(request.method())) {
                sendHttpResponse(channelHandlerContext, request, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
                ReferenceCountUtil.release(object);
                return;
            }
            final WebsocketMsgHandler handler;
            try {
                QueryStringDecoder decoder = new QueryStringDecoder(request.uri());
                if (log.isDebugEnabled()) {
                    log.debug("ws connection request : path=[{}] query=[{}]", decoder.path(), decoder.parameters());
                }
                Websocket websocket = websocketServer.mapping(decoder.path());
                if (websocket == null) {
                    log.error("ws连接时匹配错误,路径未匹配 [{}]", decoder.path());
                    sendHttpResponse(channelHandlerContext, request, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
                    ReferenceCountUtil.release(object);
                    return;
                }

                Map<String, String> headers = parseHeaders(request);
                handler = websocketServer.mappingHandler(websocket);
                handler.onOpen(channelHandlerContext.channel(), headers, decoder.parameters());
            } catch (Exception ex) {
                log.error("ws处理链接过程中异常: msg=[{}]", ex.getMessage(), ex);
                sendHttpResponse(channelHandlerContext, request, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
                ReferenceCountUtil.release(object);
                return;
            }

            WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(request.uri(), null, false);
            WebSocketServerHandshaker handshaker = wsFactory.newHandshaker(request);
            if (handshaker == null) {
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(channelHandlerContext.channel());
                ReferenceCountUtil.release(object);
                return;
            }
            handshaker.handshake(channelHandlerContext.channel(), request).addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future) throws Exception {
                    channelHandlerContext.channel().pipeline().addLast(websocketServer.getWorkThreadPool() ,handler);
                }
            });
        }

        if (object instanceof WebSocketFrame) {
            channelHandlerContext.fireChannelRead(object);
        }else{
            ReferenceCountUtil.release(object);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (cause instanceof WebSocketHandshakeException) {
            FullHttpResponse response = new DefaultFullHttpResponse(
                HTTP_1_1, HttpResponseStatus.BAD_REQUEST, Unpooled.wrappedBuffer(cause.getMessage().getBytes()));
            ctx.channel().writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
        } else {
            ctx.fireExceptionCaught(cause);
        }
    }

    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest request, FullHttpResponse response) {
        ChannelFuture channelFuture = ctx.channel().writeAndFlush(response);
        if (HttpUtil.isKeepAlive(request) || response.status().code() != HttpResponseStatus.OK.code()) {
            channelFuture.addListener(ChannelFutureListener.CLOSE);
        }
    }
}
